package com.bigdata.flink.datastream;

import com.bigdata.flink.util.StreamEnvUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

// 流联合，类型必须一致
public class UnionStreamTest {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();

    DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3);
    DataStreamSource<Integer> source6 = env.fromElements(1, 2, 3);
    DataStreamSource<Integer> source2 = env.fromElements(11, 22, 33);
    DataStreamSource<Integer> source4 = env.fromElements(11, 22, 33);
    DataStreamSource<Integer> source5 = env.fromElements(11, 22, 33);
    DataStreamSource<String> source3 = env.fromElements("111", "222", "333", "44444");

    DataStream<Integer> union = source1.union(source6, source2, source4).union(source5).union(source3.map(Integer::parseInt));

    union.print();
    env.execute();
  }
}
